package com.doit.day02;

import com.alibaba.fastjson.JSON;
import com.doit.day01.MyConfig;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class _01_练习之生产者 {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, MyConfig.HOST_AND_PORT);
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        props.setProperty(ProducerConfig.ACKS_CONFIG,"-1");

        //创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        EventLog eventLog = new EventLog();

        while (true){
            //给event的各个参数设置值
            eventLog.setGuid(RandomUtils.nextInt(0,1000000));
            eventLog.setEventId(RandomStringUtils.randomAlphabetic(10));
            eventLog.setTimestamp(System.currentTimeMillis());
            //将对象转变成json串
            String jsonString = JSON.toJSONString(eventLog);
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("event-log", jsonString);
            //写入数据
            producer.send(producerRecord);
            producer.flush();
            Thread.sleep(RandomUtils.nextInt(10,100));
        }



    }
}
